package com.demo.util;

import com.alibaba.fastjson.JSONObject;
import com.demo.result.Result;
import com.demo.result.ResultUnite;
import com.demo.exception.GlobalException;
import com.demo.result.ResultReturn;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

@Component
@Slf4j
public class RabbitUtils {

    @Resource
    RabbitTemplate rabbitTemplate;

    @Resource
    GlobalKey globalKey;

    // =========================================== 统一工作模式 ===========================================

    public <K, V> Result<ResultUnite> sendWorK(String routingKey, Map<K, V> map) throws Exception {
        return sendMessage(globalKey.getDirectExchange(), routingKey, map);
    }

    public <T> Result<ResultUnite> sendWorK(String routingKey, T tClass) throws Exception {
        return sendMessage(globalKey.getDirectExchange(), routingKey, tClass);
    }

    // =========================================== 统一扇出模式 ===========================================

    public <K, V> Result<ResultUnite> sendFount(String routingKey, Map<K, V> map) throws Exception {
        return sendMessage(globalKey.getFountExchange(), routingKey, map);
    }

    public <T> Result<ResultUnite> sendFount(String routingKey, T tClass) throws Exception {
        return sendMessage(globalKey.getFountExchange(), routingKey, tClass);
    }

    // =========================================== 统一路由模式 ===========================================

    public <K, V> Result<ResultUnite> sendHeaders(String routingKey, Map<K, V> map) throws Exception {
        return sendMessage(globalKey.getHeadersExchange(), routingKey, map);
    }

    public <T> Result<ResultUnite> sendHeaders(String routingKey, T tClass) throws Exception {
        return sendMessage(globalKey.getHeadersExchange(), routingKey, tClass);
    }

    // =========================================== 统一主题模式 ===========================================
    public <K, V> Result<ResultUnite> sendTopic(String routingKey, Map<K, V> map) throws Exception {
        return sendMessage(globalKey.getTopicExchange(), routingKey, map);
    }

    public <T> Result<ResultUnite> sendTopic(String routingKey, T tClass) throws Exception {
        return sendMessage(globalKey.getTopicExchange(), routingKey, tClass);
    }

    // =========================================== 所有模式 ===========================================

    public <K, V> Result<ResultUnite> sendMessage(String exchange, String routingKey, Map<K, V> map) throws Exception {
        return sendMessage(exchange, routingKey, map, null);
    }

    public <T> Result<ResultUnite> sendMessage(String exchange, String routingKey, T tClass) throws Exception {
        return sendMessage(exchange, routingKey, tClass, null);
    }


    public <K, V> Result<ResultUnite> sendMessage(String exchange, String routingKey, Map<K, V> map, MessagePostProcessor messagePostProcessor) throws Exception {
        return sendMessage(exchange, routingKey, map, messagePostProcessor, null);
    }

    public <T> Result<ResultUnite> sendMessage(String exchange, String routingKey, T tClass, MessagePostProcessor messagePostProcessor) throws Exception {
        return sendMessage(exchange, routingKey, tClass, messagePostProcessor, null);
    }


    /**
     * 发送统一所有模式
     *
     * @param map                  消息体
     * @param messagePostProcessor 消息额外参数
     * @param correlationData      消息唯一标识
     * @return 200
     * @throws Exception AmqpException
     */
    public <K, V> Result<ResultUnite> sendMessage(String exchange, String routingKey, Map<K, V> map, MessagePostProcessor messagePostProcessor, CorrelationData correlationData) throws Exception {
        try {
            if (messagePostProcessor != null && correlationData != null) {
                rabbitTemplate.convertAndSend(exchange, routingKey, map, messagePostProcessor, correlationData);
            } else if (messagePostProcessor != null) {
                rabbitTemplate.convertAndSend(exchange, routingKey, map, messagePostProcessor);
            } else {
                rabbitTemplate.convertAndSend(exchange, routingKey, map);
            }
            return ResultReturn.success();
        } catch (AmqpException amqpException) {
            GlobalException.rabbitMq(amqpException, "发送消息失败，" +
                    "路由：" + exchange +
                    "routingKey" + routingKey +
                    "参数类型 integer 消息体：" + map.toString());
        }
        return ResultReturn.failError("发送消息失败");
    }

    public <T> Result<ResultUnite> sendMessage(String exchange, String routingKey, T tClass, MessagePostProcessor messagePostProcessor, CorrelationData correlationData) throws Exception {
        try {
            if (messagePostProcessor != null && correlationData != null) {
                rabbitTemplate.convertAndSend(exchange, routingKey, tClass, messagePostProcessor, correlationData);
            } else if (messagePostProcessor != null) {
                rabbitTemplate.convertAndSend(exchange, routingKey, tClass, messagePostProcessor);
            } else {
                rabbitTemplate.convertAndSend(exchange, routingKey, tClass);
            }
            return ResultReturn.success();
        } catch (AmqpException amqpException) {
            GlobalException.rabbitMq(amqpException, "发送消息失败，" +
                    "路由：" + exchange +
                    "routingKey" + routingKey +
                    "消息体：" + JSONObject.toJSONString(tClass));
        }
        return ResultReturn.failError("发送消息失败");
    }

    /**
     * 延时消息【通过延时交换机插件实现】
     */
    public Result<ResultUnite> sendDelayMsg(String exchange, String queue, String routingKey, long time, JSONObject jsonObject) throws Exception {
        try {
            MessagePostProcessor message = (Message msg) -> {
                msg.getMessageProperties().setHeader("x-delay", time);
                return msg;
            };
            rabbitTemplate.convertAndSend(globalKey.getDelayExchange(), routingKey, jsonObject, message);
            log.info("*******交换机：" + exchange);
            log.info("*******队列名：" + queue);
            log.info("*******RountingKey：" + routingKey);
            return ResultReturn.success();
        } catch (AmqpException amqpException) {
            GlobalException.rabbitMq(amqpException, "发送消息失败," +
                    "交换机：" + exchange +
                    "routingKey" + routingKey +
                    "消息体：" + JSONObject.toJSONString(jsonObject));
        }
        return ResultReturn.failError("发送消息失败");
    }

}

